package com.th3t.enhance.akka.tutorial.actor;

import java.util.concurrent.TimeUnit;

import scala.concurrent.duration.Duration;
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.routing.RoundRobinPool;

import com.th3t.enhance.akka.tutorial.msg.CalculateMsg;
import com.th3t.enhance.akka.tutorial.msg.PiApproximationMsg;
import com.th3t.enhance.akka.tutorial.msg.ResultMsg;
import com.th3t.enhance.akka.tutorial.msg.WorkMsg;

public class Master extends UntypedActor {
	private final int nrOfMessages;
	private final int nrOfElements;
	
	private double pi;
	private int nrOfResults;
	private final long start = System.currentTimeMillis();

	private final ActorRef listener;
	private final ActorRef workerRouter;

	public Master(final int nrOfWorkers, int nrOfMessages, int nrOfElements,
			ActorRef listener) {
		this.nrOfMessages = nrOfMessages;
		this.nrOfElements = nrOfElements;
		this.listener = listener;

		workerRouter = this.getContext().actorOf(
				Props.create(Worker.class).withRouter(
						new RoundRobinPool(nrOfWorkers)), "workerRouterPool");
	}

	public void onReceive(Object message) {
		// handle messages ...
		if (message instanceof CalculateMsg) {
			for (int start = 0; start < nrOfMessages; start++) {
				workerRouter.tell(new WorkMsg(start, nrOfElements), getSelf());
			}
		} else if (message instanceof ResultMsg) {
			ResultMsg result = (ResultMsg) message;
			pi += result.getValue();
			nrOfResults += 1;
			if (nrOfResults == nrOfMessages) {
				// Send the result to the listener
				Duration duration = Duration.create(System.currentTimeMillis()
						- start, TimeUnit.MILLISECONDS);
				listener.tell(new PiApproximationMsg(pi, duration), getSelf());
				// Stops this actor and all its supervised children
				getContext().stop(getSelf());
			}
		} else {
			unhandled(message);
		}
	}

	public static Props props(final int nrOfWorkers, int nrOfMessages,
			int nrOfElements, ActorRef listener) {
		return Props.create(Master.class, nrOfWorkers, nrOfMessages,
				nrOfElements, listener);
	}
}